import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.junit.Test;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * git地址

 https://gitee.com/yujianlong1988/java8andDisruptor.git
 * 日期类
 * @author yujianlong
 * @create 2018-06-16 11:29
 **/
public class DisruptorTest1 {

	Logger logger = Logger.getLogger("dis");

	public static class TBean {
		String name;
		String sex;
		String oid;
		String joid;
		Long cou;

		public TBean() {
		}

		public TBean(String name, String sex, String oid, String joid, Long cou) {
			this.name = name;
			this.sex = sex;
			this.oid = oid;
			this.joid = joid;
			this.cou = cou;
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}

		public String getSex() {
			return sex;
		}

		public void setSex(String sex) {
			this.sex = sex;
		}

		public String getOid() {
			return oid;
		}

		public void setOid(String oid) {
			this.oid = oid;
		}

		public String getJoid() {
			return joid;
		}

		public void setJoid(String joid) {
			this.joid = joid;
		}

		public Long getCou() {
			return cou;
		}

		public void setCou(Long cou) {
			this.cou = cou;
		}

		@Override
		public String toString() {
			return "TBean{" +
					"name='" + name + '\'' +
					", sex='" + sex + '\'' +
					", oid='" + oid + '\'' +
					", joid='" + joid + '\'' +
					", cou=" + cou +
					'}';
		}
	}

	public static class Tevent {
		private String value;
		private LongAdder countAdder = new LongAdder();
		private Map<String, String> value1 = new HashMap<>();

		private List<TBean> tbeans = new ArrayList<>();

		public Tevent() {
//			this.getCountAdder().add(0);
		}

		public Tevent(String value, List<TBean> tBeans) {
			this.value = value;
			this.tbeans = tBeans;
		}

		public Tevent(String value, Map<String, String> value1, List<TBean> tbeans) {
			this.value = value;
			this.value1 = value1;
			this.tbeans = tbeans;
		}

		public String getValue() {
			return value;
		}

		public void setValue(String value) {
			this.value = value;
		}

		public Map<String, String> getValue1() {
			return value1;
		}

		public void setValue1(Map<String, String> value1) {
			this.value1 = value1;
		}

		public LongAdder getCountAdder() {
			return countAdder;
		}

		public void setCountAdder(LongAdder countAdder) {
			this.countAdder = countAdder;
		}

		@Override
		public String toString() {
			return "Tevent{" +
					"value='" + value + '\'' +
					", countAdder=" + countAdder +
					", value1=" + value1 +
					", tbeans=" + tbeans +
					'}';
		}

		public List<TBean> getTbeans() {
			return tbeans;
		}

		public void setTbeans(List<TBean> tbeans) {
			this.tbeans = tbeans;
		}
	}
	@Test
	public void test1() {//菱形
		final int DEFAULT_CORE_THREADS = Runtime.getRuntime().availableProcessors();
		ThreadFactory threadFactory = Executors.defaultThreadFactory();
		int bufferSize = 4;
		Disruptor<Tevent> disruptor =
				new Disruptor<Tevent>(Tevent::new, bufferSize, threadFactory, ProducerType.MULTI, new SleepingWaitStrategy());
		disruptor.handleExceptionsWith(new ExceptionHandler<Tevent>() {

			@Override
			public void handleEventException(Throwable throwable, long l, Tevent tevent) {
				throwable.printStackTrace();
				logger.log(Level.WARNING,"异常数据节点:{}", tevent);
				logger.log(Level.WARNING,"异常原因:{}", throwable.getMessage());
			}

			@Override
			public void handleOnStartException(Throwable throwable) {
				logger.log(Level.WARNING,"handleOnStartException:{}", throwable);
			}

			@Override
			public void handleOnShutdownException(Throwable throwable) {
				logger.log(Level.WARNING,"handleOnShutdownException:{}", throwable);
			}
		});

		EventHandler<Tevent> e0 = (Tevent tevent, long var2, boolean var4) -> {
			List<TBean> beans = IntStream.rangeClosed(0, 100).boxed().map(t -> {
				TBean b = new TBean();
				b.setName("" + t);
				b.setOid(UUID.randomUUID().toString().replace("-",""));
				return b;
			}).collect(Collectors.toList());
			tevent.getTbeans().addAll(beans);

		};

		EventHandler<Tevent> e11 = (Tevent tevent, long var2, boolean var4) -> {
			tevent.getTbeans().parallelStream().forEach(b -> {
				try {
					TimeUnit.MILLISECONDS.sleep(1);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				b.setSex("nnn1");
			});
		};


		EventHandler<Tevent> e2 = (Tevent tevent, long var2, boolean var4) -> {
			tevent.getTbeans().parallelStream().forEach(b -> {
				try {
					TimeUnit.MILLISECONDS.sleep(1);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				b.setJoid("aos1");
			});
		};

		EventHandler<Tevent> e3 = (Tevent tevent, long var2, boolean var4) -> {
			System.out.println("tevent.getValue1()===" + tevent.getValue1());
			Map<String, String> hh = new HashMap<>();
			hh.put("aa", "bb");
			tevent.getCountAdder().add(99);
			tevent.getValue1().putAll(hh);
		};
		disruptor.handleEventsWith(e0).handleEventsWith(e11, e2).handleEventsWith(e3);


		disruptor.start();
			/*定义单参发射器*/
		RingBuffer<Tevent> ringBuffer = disruptor.getRingBuffer();
		EventTranslatorOneArg<Tevent, Tevent> TRANSLATOR_ONE = (dayCutTransferEvent1, sequence, dayCutTransferEvent2) -> {
			dayCutTransferEvent1.setValue(dayCutTransferEvent2.getValue());

			dayCutTransferEvent1.setCountAdder(dayCutTransferEvent2.getCountAdder());
			dayCutTransferEvent1.setValue1(dayCutTransferEvent2.getValue1());
			dayCutTransferEvent1.setTbeans(dayCutTransferEvent2.getTbeans());
		};


		Tevent e1 = new Tevent();
		e1.setValue("kk11");
		ringBuffer.publishEvent(TRANSLATOR_ONE, e1);
		disruptor.shutdown();
		long end = System.currentTimeMillis();
//		System.out.println("耗时=" + (end - start));
		System.out.println(e1);
//		logger.info(Level.WARNING,"kkkk","aa");
		logger.log(Level.WARNING,"异常数据节点:{0}", "aa");
		logger.log(Level.INFO,"异常数据节点1:{0}", "aa");

	}




}
